package com.yifeng.repo.controller.kafka.message.handler;

import com.gomcarter.frameworks.base.exception.CustomException;
import com.google.common.base.Strings;
import com.yifeng.repo.controller.kafka.configure.KafkaProperties;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.security.scram.ScramLoginModule;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Created by daibing on 2024/5/6.
 */
public class KafkaHelper {


    public static String getAuthContent(String username, String password) {
        return String.format("%s required username=\"%s\" password=\"%s\";", ScramLoginModule.class.getName(), username, password);
    }

    public static String getProducerAuthContent(KafkaProperties.TopicProperties topicProperties) {
        if (!Strings.isNullOrEmpty(topicProperties.getProducerUsername()) || !Strings.isNullOrEmpty(topicProperties.getProducerPassword())) {
            return getAuthContent(topicProperties.getProducerUsername(), topicProperties.getProducerPassword());
        } else {
            return getAuthContent(topicProperties.getUsername(), topicProperties.getPassword());
        }
    }

    public static String getConsumerAuthContent(KafkaProperties.TopicProperties topicProperties) {
        if (!Strings.isNullOrEmpty(topicProperties.getConsumerUsername()) || !Strings.isNullOrEmpty(topicProperties.getConsumerPassword())) {
            return getAuthContent(topicProperties.getConsumerUsername(), topicProperties.getConsumerPassword());
        } else {
            return getAuthContent(topicProperties.getUsername(), topicProperties.getPassword());
        }
    }

    public static KafkaProperties.TopicProperties getTopicProperties(KafkaProperties properties, String topic) {
        return properties.getTopicPropertiesList()
                .stream()
                .filter(s -> s.getTopic().equals(topic))
                .findFirst()
                .orElseThrow(() -> new CustomException("没有Topic配置：" + topic));
    }

    public static String buildPartitionLog(Collection<TopicPartition> partitions) {
        Map<String, List<TopicPartition>> topic2Partitions = partitions.stream().collect(Collectors.groupingBy(TopicPartition::topic));
        StringBuilder sb = new StringBuilder();
        topic2Partitions.forEach((topic, partitionList) -> {
            String ids = partitionList.stream()
                    .map(topicPartition -> String.valueOf(topicPartition.partition()))
                    .reduce((a, b) -> a + "," + b)
                    .orElse("NONE");
            sb.append("topic=").append(topic).append(",");
            sb.append("partitions=").append(ids).append(",");
        });
        return sb.toString();
    }

}
